package com.violence.web.service;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpHost;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.ActionType;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
import org.elasticsearch.action.bulk.*;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.*;
import org.elasticsearch.action.support.IndicesOptions;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.*;
import org.elasticsearch.common.Nullable;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.GeoBoundingBoxQueryBuilder;
import org.elasticsearch.index.query.GeoDistanceQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.*;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * @author liuf
 * @date 2020-09-01 12:00:00
 * elasticsearch 基础工具类
 */
@Component
public class BaseElasticService {
    private static final Logger log = LoggerFactory.getLogger(BaseElasticService.class);

    @Autowired
    private RestHighLevelClient restHighLevelClient;

    String child [] ={"send_log","event_log","lable_log"};
    /**
     * 普通 创建索引
     * @param idxName
     */
    public void createIndex(String idxName){
        try {
            if (indexExist(idxName)) {
                log.error("idxName:{} already exists", idxName);
                return;
            }

            CreateIndexRequest request = new CreateIndexRequest(idxName);
            buildSetting(request);
            XContentBuilder  mappingBuilder = XContentFactory.jsonBuilder()
                    .startObject()
                    .startObject("properties")
                    .startObject("type").field("type", "keyword").endObject()
                    .startObject("email").field("type", "long").endObject()
                    .startObject("mobile").field("type", "boolean").endObject()
                    // 时间类型
                    .startObject("create_time").field("type", "date").field("format", "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis").endObject()
                    // 子类处理
                    .startObject("relation_type")
                    .field("type", "join")
                    .field("eager_global_ordinals", "true")
                    .field("relations").startObject().array("parent",child).endObject()
                    .endObject()

                    .endObject()
                    .endObject();
            request.mapping(mappingBuilder);
            CreateIndexResponse res = restHighLevelClient.indices().create(request, RequestOptions.DEFAULT);
            if (!res.isAcknowledged()) {
                log.error("createIndex  failed");
                throw new RuntimeException("createIndex  failed");
            }
        } catch (Exception e) {
            log.error(e.getMessage(),e);
        }
    }

    /**
     * 普通 创建索引
     *
     * @param idxName
     * @param idxSQL
     */
    public void createIndex(String idxName,String idxSql){
        try {
            if (indexExist(idxName)) {
                log.error("idxName:{} already exists,idxSql:{}", idxName,idxSql);
                return;
            }
            CreateIndexRequest request = new CreateIndexRequest(idxName);
            buildSetting(request);
            request.mapping(idxSql,XContentType.JSON);
            CreateIndexResponse res = restHighLevelClient.indices().create(request, RequestOptions.DEFAULT);
            if (!res.isAcknowledged()) {
                log.error("createIndex  failed");
                throw new RuntimeException("createIndex  failed");
            }
        } catch (Exception e) {
            log.error(e.getMessage(),e);
        }
    }



    /**
     * 创建索引 设置分片
     * @param idxName  索引
     * @param numberOfShards 主片
     * @param numberOfReplicas 副片
     */
    public void createIndex(String idxName,Long numberOfShards,Long numberOfReplicas){
        try {
            if (indexExist(idxName)) {
                log.error("idxName:{} already exists,idxSql:{}", idxName);
                return;
            }
            CreateIndexRequest request = new CreateIndexRequest(idxName);
            buildSetting(request,numberOfShards,numberOfReplicas);
            CreateIndexResponse res = restHighLevelClient.indices().create(request, RequestOptions.DEFAULT);
            if (!res.isAcknowledged()) {
                log.error("createIndex  failed");
                throw new RuntimeException("初始化失败");
            }
        } catch (Exception e) {
            log.error(e.getMessage(),e);
        }
    }

    /**
     * 普通 更新索引
     *
     * @param idxName    索引
     * @param field      字段
     * @param fieldType  字段类型
     */
    public void  putMapping(String idxName,String field,String fieldType){
        try {
            PutMappingRequest  request = new PutMappingRequest(idxName);
            JSONObject jsonObject = new JSONObject();
            JSONObject properties = new  JSONObject();
            properties.put(field,mappingType(fieldType));
            jsonObject.put("properties",properties);
            request.source(jsonObject.toJSONString(), XContentType.JSON);
            AcknowledgedResponse res = restHighLevelClient.indices().putMapping(request, RequestOptions.DEFAULT);
            if (!res.isAcknowledged()) {
                log.error("putMapping  failed");
                throw new RuntimeException("初始化失败");
            }
        } catch (Exception e) {
            log.error(e.getMessage(),e);
        }
    }

    /**
     * 设置字段类型
     * @param fieldType
     * @return
     */
    public JSONObject mappingType(String fieldType) {
        JSONObject type =  new JSONObject();
        if ("date".equals(fieldType)) {
            type.put("format","yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis");
            type.put("type","date");
        } else if ("string".equals(fieldType)) {
            type.put("type","keyword");
        }  else if ("text".equals(fieldType)) {
            JSONObject keyword = new JSONObject();
            keyword.put("type","keyword");
            keyword.put("ignore_above",356);
            type.put("fields",keyword);
            type.put("type","text");
        } else if ("long".equals(fieldType)) {
            type.put("type","long");
        } else if ("double".equals(fieldType)) {
            type.put("type","double");
        } else if ("boolean".equals(fieldType)) {
            type.put("type","boolean");
        } else if ("join".equals(fieldType)) {
            type.put("type","join");
            type.put("eager_global_ordinals", true);
            JSONObject relations =  new JSONObject();
            relations.put("parent",child);
            type.put("relations",relations);
        } else {
            type.put("type",fieldType);
        }
        return type;
    }


    /**
     * 默认索引
     * @param request
     */
    public void buildSetting(CreateIndexRequest request) {
        //可以自定义分片、副本的数量   默认  分片5  副本 1
        request.settings(Settings.builder().put("index.number_of_shards",1).put("index.number_of_replicas",1));
    }

    /**
     * 设置索引信息
     * @param request
     * @param numberOfShards
     * @param numberOfReplicas
     */
    public void buildSetting(CreateIndexRequest request,Long numberOfShards,Long numberOfReplicas) {
        //可以自定义分片、副本的数量   默认  分片5  副本 1
        request.settings(Settings.builder().put("index.number_of_shards",numberOfShards).put("index.number_of_replicas",numberOfReplicas));
    }

    /**
     * 判断索引是否存在
     *
     * @param idxName
     * @return
     * @throws Exception
     */
    public boolean indexExist(String idxName) {
        boolean exists = false;
        try {
            GetIndexRequest request = new GetIndexRequest();
            //TRUE-返回本地信息检索状态，FALSE-还是从主节点检索状态
            request.local(false);
            //是否适应被人可读的格式返回
            request.humanReadable(true);
            //是否为每个索引返回所有默认设置
            request.includeDefaults(false);
            request.indices(idxName);
            //控制如何解决不可用的索引以及如何扩展通配符表达式,忽略不可用索引的索引选项，仅将通配符扩展为开放索引，并且不允许从通配符表达式解析任何索引
//            request.indicesOptions(IndicesOptions.lenientExpandOpen());
            exists = restHighLevelClient.indices().exists(request, RequestOptions.DEFAULT);
            log.error("{} index  exists",idxName);
        } catch (IOException e) {
            log.error("index exist  failed");
            log.error(e.toString(),e);
        }
        return exists;
    }

    /**
     * 删除索引
     * 禁止使用方法
     * @param idxName
     */
    public boolean deleteIndex(String idxName) {
        boolean acknowledged = false;
        try {
            DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(idxName);
            deleteIndexRequest.indicesOptions(IndicesOptions.LENIENT_EXPAND_OPEN);
            AcknowledgedResponse delete = restHighLevelClient.indices().delete(deleteIndexRequest, RequestOptions.DEFAULT);
            acknowledged = delete.isAcknowledged();
        } catch (IOException e) {
            log.error(e.getMessage(),e);
        }
        return acknowledged;
    }
    /**
     * 创建模版
     */
    @Test
    public  void createIndexTemplate() {
        try {
            PutIndexTemplateRequest request = new PutIndexTemplateRequest("contact_index_template");
            request.order(1);
            request.version(1);
            // 设置模版 匹配
            List<String> indexPatterns = new ArrayList<String>();
            indexPatterns.add("contact_*");
            request.patterns(indexPatterns);
            // 设置 mapping
            // 配置字段信息
            JSONObject fieldList =  new JSONObject();
            fieldList.put("firstName","keyword");
            fieldList.put("isAlive","boolean");
            fieldList.put("create_time","date");

            JSONObject properties =  new JSONObject();
            JSONObject field =  new JSONObject();
            if (fieldList != null && fieldList.keySet().size() !=0) {
                for (String key :fieldList.keySet()) {
                    field.put(key,mappingType(fieldList.getString(key)));
                }
            }
            // 设置 特殊关系处理
            if (true) {
                field.put("relation_type",mappingType("join"));
            }
            properties.put("properties",field);
            request.mapping(properties.toJSONString(),XContentType.JSON);
            // 设置 setting
            Map<String, Object> settings = new HashMap<>();
            settings.put("number_of_shards", 1);
            settings.put("number_of_replicas", 1);
            request.settings(settings);
            AcknowledgedResponse response =  restHighLevelClient.indices().putTemplate(request, RequestOptions.DEFAULT);
            if (response.isAcknowledged()) {
                log.info("创建模版成功!");
            } else {
                log.error("创建模版失败!");
            }
        } catch (Exception e) {
            // TODO: handle exception
            log.error("创建模版失败!");
            log.error(e.getMessage(),e);
        } finally {
            try {
                restHighLevelClient.close();
            } catch (Exception e) {
                log.error(e.toString(),e);
            }
        }
    }




    /**
     * 获取模版
     */
    @Test
    public  void getIndexTempTemplate() {
        try {
            GetIndexTemplatesRequest request = new GetIndexTemplatesRequest("contact_index_template");
            GetIndexTemplatesResponse response = restHighLevelClient.indices().getIndexTemplate(request, RequestOptions.DEFAULT);
            List<IndexTemplateMetaData> indexTemplateMetaDataList = response.getIndexTemplates();
            if (indexTemplateMetaDataList.size()!=0) {
                for (IndexTemplateMetaData templateMetaData : indexTemplateMetaDataList) {
//                    templateMetaData.order();
//                    templateMetaData.mappings();
//                    templateMetaData.settings();
//                    templateMetaData.patterns();
//                    templateMetaData.version();
                    log.info("模版详情 templateMetaData:{} mappings:{}", templateMetaData.name(),templateMetaData.mappings().getSourceAsMap());
                }
                log.info("模版已存在!");
            } else {
                log.info("模版不存在!");
            }
        } catch (Exception e) {
            // TODO: handle exception
            log.error(e.toString(),e);
        } finally {
            try {
                restHighLevelClient.close();
            } catch (Exception e) {
                log.error(e.toString(),e);
            }
        }
    }

    /**
     * 创建模版
     */
    public boolean indexExistTempTemplate() {
        boolean exist = false;
        try {
            GetIndexTemplatesRequest request = new GetIndexTemplatesRequest("contact_index_template");
            GetIndexTemplatesResponse response = restHighLevelClient.indices().getIndexTemplate(request, RequestOptions.DEFAULT);
            List<IndexTemplateMetaData> indexTemplateMetaDataList = response.getIndexTemplates();
            if (indexTemplateMetaDataList.size()!=0) {
                for (IndexTemplateMetaData templateMetaData:indexTemplateMetaDataList) {
//                    templateMetaData.order();
//                    templateMetaData.mappings();
//                    templateMetaData.settings();
//                    templateMetaData.patterns();
//                    templateMetaData.version();
                    log.info("模版详情 templateMetaData:{} mappings:{}", templateMetaData.name(),templateMetaData.mappings().getSourceAsMap());
                }
                exist = true;
                log.info("模版已存在!");
            } else {
                log.info("模版不存在!");
            }
        } catch (Exception e) {
            // TODO: handle exception
            log.error(e.toString(),e);
        } finally {
            try {
                restHighLevelClient.close();
            } catch (Exception e) {
                log.error(e.toString(),e);
            }
        }
        return exist;
    }

    /**
     * 单独保存
     *
     * @param indexName
     * @return
     */
    public void insert(String indexName, JSONObject jsonObject) {
        IndexRequest indexRequest = new IndexRequest(indexName);
        indexRequest.source(jsonObject.toJSONString(), XContentType.JSON);
        try {
            IndexResponse indexResponse = restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
            log.info("新增文档成功!");
        } catch (IOException e) {
            log.info(e.toString(),e);

        }
    }
    /**
     * 单独保存
     *
     * @param indexName
     * @return
     */
    public void insertChild(String indexName) {
        IndexRequest indexRequest = new IndexRequest(indexName);
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("email", "11111@qq.com");
        jsonObject.put("mobile", "111111");
        jsonObject.put("type", 1);
        jsonObject.put("name", "11111");
        indexRequest.source(jsonObject.toJSONString(), XContentType.JSON);
        indexRequest.routing("2");

        try {
            IndexResponse indexResponse = restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
            if (indexResponse != null) {
                String id = indexResponse.getId();
                String index = indexResponse.getIndex();
                long version = indexResponse.getVersion();
                log.info("index:{},id:{}", index, id);
                if (indexResponse.getResult() == DocWriteResponse.Result.CREATED) {
                    log.info("新增文档成功!" + index + "-" + id + "-" + version);
                } else if (indexResponse.getResult() == DocWriteResponse.Result.UPDATED){
                    log.info("修改文档成功!");
                }

                ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
                if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
                    log.info("分片处理信息.....");
                }
                if (shardInfo.getFailed() > 0) {
                    for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                        String reason = failure.reason();
                        log.error("副本失败原因：" + reason);
                    }
                }

            }
        } catch (Exception e) {
            log.error(e.getMessage());
        } finally {
            try {
                restHighLevelClient.close();
            } catch (Exception e) {
                log.error(e.toString(),e);
            }
        }
    }

    /**
     * 单独保存
     * @param indexName
     * @param type
     * @param parent
     */
    public void insertParent(String indexName,String type,String parent) {
        IndexRequest indexRequest = new IndexRequest(indexName);
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("email", "1111111@qq.com");
        jsonObject.put("mobile", "222222");
        jsonObject.put("type", 1);
        jsonObject.put("name", "violence");
        JSONObject object = new JSONObject();
        object.put("name",type);
        if (StringUtils.isNotBlank(parent)){
            object.put("parent",parent);
        }
        jsonObject.put("relation_type",object);
        indexRequest.source(jsonObject.toJSONString(), XContentType.JSON);
        if (StringUtils.isNotBlank(parent)){
            indexRequest.routing(parent);
        }
        try {
            IndexResponse indexResponse = restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
            if (indexResponse != null) {
                String id = indexResponse.getId();
                String index = indexResponse.getIndex();
                long version = indexResponse.getVersion();
                if (indexResponse.getResult() == DocWriteResponse.Result.CREATED) {
                    log.info("新增文档成功!" + index + "-" + id + "-" + version);
                } else if (indexResponse.getResult() == DocWriteResponse.Result.UPDATED){
                    log.info("修改文档成功!");
                }
                ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
                if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
                    log.info("分片处理信息.....");
                }
                if (shardInfo.getFailed() > 0) {
                    for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                        String reason = failure.reason();
                        log.error("副本失败原因：" + reason);
                    }
                }
            }
        } catch (Exception e) {
            log.error(e.toString(),e);
        } finally {
            try {
                restHighLevelClient.close();
            } catch (Exception e) {
                log.error(e.toString(),e);
            }
        }
    }



    /**
     * 异步方法
     * @return
     * @throws IOException
     */
    public void  indexAsync() throws IOException, InterruptedException {
        ActionListener listener = new ActionListener<IndexResponse>() {
            @Override
            public void onResponse(IndexResponse indexResponse) {
                log.info("Async:" + indexResponse.status().name());
                if (indexResponse.getResult() == DocWriteResponse.Result.CREATED) {
                    // Todo
                } else if (indexResponse.getResult() == DocWriteResponse.Result.UPDATED) {
                    // Todo
                }
                // 处理成功分片小于总分片的情况
                ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
                if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
                    // Todo
                }
            }

            @Override
            public void onFailure(Exception e) {
                log.info("AsyncFailure:" + e.getMessage());
            }
        };

        IndexRequest indexRequest = new IndexRequest("twitter")
                .source("user", "kimchy",
                        "postDate", new Date(),
                        "message", "trying out Elasticsearch")
                .routing("my_route");   // 指定路由
        restHighLevelClient.indexAsync(indexRequest, RequestOptions.DEFAULT, listener);  // 异步方式
    }

    /**
     * 批量处理文档
     * @param list 需要处理的文档列表
     */
    public void createMultiDoc(List<Object> list){
        BulkRequest request = new BulkRequest();
        /**
         * 遍历
         * IndexRequest不设置id，让系统自己添加
         * DeleteRequest 批量删除
         * UpdateRequest 批量更新
         */
        list.forEach((obj)->{
            request.add(new IndexRequest(obj.getClass().getSimpleName().toLowerCase())
                    .source(JSONObject.toJSONString(obj), XContentType.JSON));
        });
        try {
            BulkResponse bulkResponse = restHighLevelClient.bulk(request, RequestOptions.DEFAULT);
            log.info(JSONObject.toJSONString(bulkResponse));
            bulkResponse.forEach((BulkItemResponse bulkItemResponse)->{
                DocWriteResponse itemResponse = bulkItemResponse.getResponse();
                switch (bulkItemResponse.getOpType()) {
                    case INDEX:
                    case CREATE:
                        IndexResponse indexResponse = (IndexResponse) itemResponse;
                        log.info(JSONObject.toJSONString(indexResponse));
                        if(DocWriteResponse.Result.CREATED == indexResponse.getResult()){
                            log.info("添加成功");
                        }
                        break;
                    case UPDATE:
                        UpdateResponse updateResponse = (UpdateResponse) itemResponse;
                        break;
                    case DELETE:
                        DeleteResponse deleteResponse = (DeleteResponse) itemResponse;
                        break;
                    default:break;
                }
            });
        } catch (IOException e) {
            log.error(e.toString(),e);
        } finally {
            try {
                restHighLevelClient.close();
            } catch (Exception e) {
                log.error(e.toString(),e);
            }
        }
    }

    /**
     * 批量保存
     *
     * @param indexName
     */
    public void batchInsert(String indexName) {

        BulkRequest request = new BulkRequest();

        JSONObject jsonObject = null;
        for (int i = 0; i < 1; i++) {
            jsonObject = new JSONObject();
            jsonObject.put("email", "836291658@qq.com");
            jsonObject.put("mobile", "1768698105" + i);

            JSONObject object = new JSONObject();
            object.put("name","parent");
//            if (StringUtils.isNotBlank(parent)){
//                object.put("parent",parent);
//            }
            jsonObject.put("relation_type",object);

            request.add(new IndexRequest(indexName).source(jsonObject.toJSONString(), XContentType.JSON).id(UUID.randomUUID().toString()));
        }

        try {
            BulkResponse bulkResponse = restHighLevelClient.bulk(request, RequestOptions.DEFAULT);


            if (bulkResponse.hasFailures()) {
                log.error("buildFailureMessage:{}",bulkResponse.buildFailureMessage());
                BulkItemResponse [] items =   bulkResponse.getItems();
                if (items.length !=0) {
                    for ( BulkItemResponse item:items) {
                        log.error("items:{}",item.getFailure());
                    }
                }
                log.info("失败");
            } else {
                bulkResponse.forEach((BulkItemResponse bulkItemResponse)->{
                    DocWriteResponse itemResponse = bulkItemResponse.getResponse();
                    switch (bulkItemResponse.getOpType()) {
                        case INDEX:
                        case CREATE:
                            IndexResponse indexResponse = (IndexResponse) itemResponse;
                            log.info("indexResponse:{}",JSONObject.toJSONString(indexResponse));
                            if(DocWriteResponse.Result.CREATED == indexResponse.getResult()){
                                log.info("添加成功");
                            }
                            break;
                        case UPDATE:
                            UpdateResponse updateResponse = (UpdateResponse) itemResponse;
                            log.info("updateResponse:{}",JSONObject.toJSONString(updateResponse));

                            log.info("修改成功");

                            break;
                        case DELETE:
                            DeleteResponse deleteResponse = (DeleteResponse) itemResponse;
                            log.info("deleteResponse:{}",JSONObject.toJSONString(deleteResponse));
                            log.info("删除成功");
                            break;
                        default:break;
                    }
                });

                log.info("成功");

            }
        } catch (Exception e) {
            log.error(e.toString(),e);
        }
    }
    /**
     * 删除数据
     *
     * @param indexName
     * @param dataId
     */
    public void delete(String indexName, String dataId) {
        DeleteRequest request = new DeleteRequest(indexName);
        request.id(dataId);
        try {
            DeleteResponse rsponse = restHighLevelClient.delete(request, RequestOptions.DEFAULT);
            if (rsponse != null) {
                String id = rsponse.getId();
                String index = rsponse.getIndex();
                long version = rsponse.getVersion();
                if (rsponse.getResult() == DocWriteResponse.Result.CREATED) {
                    log.info("创建成功!" + index + "-" + id + "-" + version);
                } else if (rsponse.getResult() == DocWriteResponse.Result.UPDATED){
                    log.info("修改文档成功!");
                } else if (rsponse.getResult() == DocWriteResponse.Result.DELETED) {
                    log.info("删除文档成功!");
                }
                ReplicationResponse.ShardInfo shardInfo = rsponse.getShardInfo();
                if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
                    log.info("分片处理信息.....");
                }
                if (shardInfo.getFailed() > 0) {
                    for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                        String reason = failure.reason();
                        log.error("副本失败原因：" + reason);
                    }
                }
            }
        } catch (Exception e) {
            log.error(e.getMessage());
        }
    }

    /**
     * 修改数据
     *
     * @param indexName
     * @param dataId
     */
    public void update(String indexName, String dataId) {
        UpdateRequest request = new UpdateRequest(indexName, dataId);
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("name", "violence_update_v3");
        request.doc(jsonObject);
        try {
            UpdateResponse rsponse = restHighLevelClient.update(request, RequestOptions.DEFAULT);
            if (rsponse != null) {
                String id = rsponse.getId();
                String index = rsponse.getIndex();
                long version = rsponse.getVersion();
                if (rsponse.getResult() == DocWriteResponse.Result.CREATED) {
                    log.info("创建成功!" + index + "-" + id + "-" + version);
                } else if (rsponse.getResult() == DocWriteResponse.Result.UPDATED){
                    log.info("修改文档成功!");
                } else if (rsponse.getResult() == DocWriteResponse.Result.DELETED) {
                    log.info("删除文档成功!");
                }
                ReplicationResponse.ShardInfo shardInfo = rsponse.getShardInfo();
                if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
                    log.info("分片处理信息.....");
                }
                if (shardInfo.getFailed() > 0) {
                    for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                        String reason = failure.reason();
                        log.error("副本失败原因：" + reason);
                    }
                }
            }
        } catch (Exception e) {
            log.error(e.getMessage());
        }
    }
    /**
     * 修改数据
     *
     * @param indexName
     * @param dataId
     */
    public void DeleteByQuery(String indexName) {
        BulkByScrollResponse response = new DeleteByQueryRequestBuilder((Client)restHighLevelClient,DeleteByQueryAction.INSTANCE)
                .filter(QueryBuilders.matchQuery("gender", "male"))
                .source(indexName)
                .get();
        long deleted = response.getDeleted();
    }

    /**
     * 批量更新
     * @param indexName
     */
    public void updateByQuery(String indexName) {
        UpdateByQueryRequestBuilder updateByQuery =
                new UpdateByQueryRequestBuilder((Client)restHighLevelClient, UpdateByQueryAction.INSTANCE);
        updateByQuery.source(indexName).abortOnVersionConflict(false);
        BulkByScrollResponse response = updateByQuery.get();
        if(response !=null){
        }
    }


    /**
     * 批量处理器
     */
    public void bulkProcessor() {
        BulkProcessor.Listener listener = new BulkProcessor.Listener() {
            //构建bulk listener
            @Override
            public void beforeBulk(long executionId, BulkRequest request) {
                //重写beforeBulk,在每次bulk request发出前执行,在这个方法里面可以知道在本次批量操作中有多少操作数
                int numberOfActions = request.numberOfActions();
                log.debug("Executing bulk [{}] with {} requests", executionId, numberOfActions);
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
                //重写afterBulk方法，每次批量请求结束后执行，可以在这里知道是否有错误发生。
                if (response.hasFailures()) {
                    log.warn("Bulk [{}] executed with failures", executionId);
                } else {
                    log.debug("Bulk [{}] completed in {} milliseconds", executionId, response.getTook().getMillis());
                }
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
                //重写方法，如果发生错误就会调用。
                log.error("Failed to execute bulk", failure);
            }

        };

        Runnable onClose = new Runnable() {
            @Override
            public void run() {
                try {
                    restHighLevelClient.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        };

        BulkProcessor.Builder builder =  BulkProcessor.builder((Client)restHighLevelClient, listener);//使用builder做批量操作的控制

        //在这里调用build()方法构造bulkProcessor,在底层实际上是用了bulk的异步操作
        builder.setBulkActions(2000); //执行多少次动作后刷新bulk.默认1000，-1禁用
        builder.setBulkSize(new ByteSizeValue(1L, ByteSizeUnit.MB));//执行的动作大小超过多少时，刷新bulk。默认5M，-1禁用
        builder.setConcurrentRequests(5);//最多允许多少请求同时执行。默认是1，0是只允许一个。
//                builder.setFlushInterval(TimeValue.timeValueSeconds(10L));//设置刷新bulk的时间间隔。默认是不刷新的。
        builder.setBackoffPolicy(BackoffPolicy.constantBackoff(TimeValue.timeValueSeconds(1L), 3)); //设置补偿机制参数。由于资源限制（比如线程池满），批量操作可能会失败，在这定义批量操作的重试次数。
        BulkProcessor bulkProcessor = builder.build();

        bulkProcessor.add(new IndexRequest().id("22").source(""));
        bulkProcessor.add(new UpdateRequest().id("22").doc("").index(""));
        bulkProcessor.add(new UpdateRequest().id("22").doc("").index(""));


        bulkProcessor.flush();
        bulkProcessor.close();
    }


    /**
     * 查询
     *
     * @param indexName
     * @param name
     */
    public void search(String indexName) {
        SearchRequest searchRequest = new SearchRequest(indexName);

        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        boolQueryBuilder.must(QueryBuilders.matchAllQuery());
        sourceBuilder.query(boolQueryBuilder);
        sourceBuilder.timeout(new TimeValue(60, TimeUnit.SECONDS));
        searchRequest.source(sourceBuilder);
        //沿用之前的查询方式就可以   基本都支持
        try {
            SearchResponse response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            SearchHits hits = response.getHits();
            for (SearchHit hit : hits) {
                String sourceAsString = hit.getSourceAsString();
                log.info(sourceAsString);
            }
        } catch (Exception e) {
            log.error(e.getMessage());
        }
    }
    @Test
    public void testBatchInsert() {
        System.out.print(1);
        scrollSearch("logs-*");
        System.out.print(3);
    }
    /**
     * 快照形式查询数据
     * @param name
     */
    public void scrollSearch(String indexName) {
        SearchRequest searchRequest = new SearchRequest("logs-2020.09.01");
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        String[] includes ={"message"};
        String[] excludes={};
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        boolQueryBuilder.must(QueryBuilders.queryStringQuery("【事件回收】 异常 公司id:1539,objectList").field("message"));
        boolQueryBuilder.must(QueryBuilders.termQuery("level.keyword","ERROR"));
        boolQueryBuilder.must(QueryBuilders.termQuery("logger_name.keyword","com.qdum.contact.service.job.EventRecycleJob"));
        sourceBuilder.query(boolQueryBuilder);
        sourceBuilder.fetchSource(includes,excludes);
        sourceBuilder.size(5000);
        searchRequest.source(sourceBuilder);
        searchRequest.scroll(TimeValue.timeValueMinutes(15));
        //游标
        String scrollId = null;
        //总数
        Long num = 0L;
        String title = "数据导出";
        Long total =0L;
        log.info("[{}] 开始 ",title);

        do {
            try {
                SearchResponse response = null;
                if (scrollId == null) {
                    response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
                    SearchHits hits = response.getHits();
                    total = hits.getTotalHits().value;
                    num = hits.getTotalHits().value;
                    scrollId = response.getScrollId();
                } else if (num != 0) {
                    SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
                    scrollRequest.scroll(TimeValue.timeValueMinutes(15));
                    response = restHighLevelClient.scroll(scrollRequest, RequestOptions.DEFAULT);
                }
                for (SearchHit hit : response.getHits()) {
                    String sourceAsString = hit.getSourceAsString();
                    // log.info(sourceAsString);
                    num --;
                }
                log.info("[{}] 执行中 {}/{}",title,num,total);
                if (num == 0) {
                    // todo 删除
                    clearScroll(scrollId);
                    log.warn("清空游标");
                    scrollId = null;
                    break;
                }

            } catch (Exception e) {
                log.error(e.getMessage());
                if (StringUtils.isNotBlank(scrollId)) {
                    clearScroll(scrollId);
                    scrollId = null;
                    break;
                }
            }
        } while (true);
        log.info("[{}] 结束",title);
    }

    /**
     * 清空索引游标
     * @param scrollId
     */
    public void clearScroll(String scrollId) {
        ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
        clearScrollRequest.addScrollId(scrollId);
        ClearScrollResponse clearScrollResponse = null;
        try {
            clearScrollResponse = restHighLevelClient.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
        } catch (Exception e) {
            log.error(e.toString(),e);
        }
        boolean succeeded = clearScrollResponse.isSucceeded();
        log.warn("succeeded:" + succeeded);
    }


    public void searchLocation(String indexName) {
        SearchRequest searchRequest = new SearchRequest(indexName);
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();


        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
//        boolQueryBuilder.must(termQueryBuilder);

        GeoDistanceQueryBuilder geoDistanceQueryBuilder = new GeoDistanceQueryBuilder("location");

        GeoBoundingBoxQueryBuilder geoBoundingBoxQueryBuilder = new GeoBoundingBoxQueryBuilder("location");


        geoBoundingBoxQueryBuilder.setCorners(42d, -72d, 40d, -74d);

//        geoDistanceQueryBuilder.point(40d, -70d);
//        geoDistanceQueryBuilder.distance(170, DistanceUnit.KILOMETERS);
//
//        boolQueryBuilder.filter(geoDistanceQueryBuilder);

//        boolQueryBuilder.must(geoBoundingBoxQueryBuilder);

        sourceBuilder.query(geoBoundingBoxQueryBuilder);
        sourceBuilder.timeout(new TimeValue(60, TimeUnit.SECONDS));
        searchRequest.source(sourceBuilder);

        //沿用之前的查询方式就可以   基本都支持

        try {
            SearchResponse response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);

            SearchHits hits = response.getHits();
            for (SearchHit hit : hits) {
                String sourceAsString = hit.getSourceAsString();
                log.info(sourceAsString);
            }
        } catch (Exception e) {
            log.error(e.getMessage());
        }
    }

}
